package com.niodata.dp.plugin.core;

import com.niodata.dp.api.common.DpTaskApi;
import java.io.IOException;
import java.util.Map;

class PluginServiceImpl implements PluginService {

  private DpTaskApi dpApi;
  private PluginContext pluginContext;

  public PluginServiceImpl(String rpcServer, int rpcPort, PluginContext context) {
    dpApi = new DpTaskApi(rpcServer, rpcPort);
    this.pluginContext = context;
  }

  public PluginServiceImpl(DpTaskApi client, PluginContext context) {

    this.dpApi = client;
    this.pluginContext = context;
  }

  @Override
  public void saveState(PluginState state) throws IOException {
    dpApi.saveJobState(pluginContext.getTaskName(), pluginContext.getJobName(),
          pluginContext.getScheduleId(),
          state.getValues());
  }

  @Override
  public PluginState getState(String taskName, String scheduleId, String jobName)
        throws IOException {
    PluginState pluginState = new PluginState();
    Map<String, Object> values = dpApi.getJobState(taskName, scheduleId, jobName);
    for (String key : values.keySet()) {
      pluginState.setValue(key, (String) values.get(key));
    }
    return pluginState;
  }

  @Override
  public void outputValues(PluginOutputValues outpuValues) throws IOException {
    dpApi.saveOutputParams(pluginContext.getTaskName(),
          pluginContext.getJobName(),
          pluginContext.getFlowExecId(),
          outpuValues.getValues());
  }

  @Override
  public void output(Output output) throws IOException {
    dpApi.saveOutputParams(pluginContext.getTaskName(),
        pluginContext.getJobName(),
        pluginContext.getFlowExecId(),
        output);
  }

  @Override
  public void publishEvent(Map<String, Object> message) throws IOException {
    dpApi.sendMessage(pluginContext.getTaskName(),
          pluginContext.getJobName(),
          pluginContext.getFlowExecId(),
          message);
  }

  @Override
  public void saveTableInfo(String taskName, String dsName, String dbName, String tableName,
        String tableDesc) throws IOException {
    dpApi.saveTableInfo(taskName, dsName, dbName, tableName, tableDesc);
  }

  @Override
  public void saveTableInfo(String dsName, String dbName, String tableName, String tableDesc)
        throws IOException {
    this.saveTableInfo(pluginContext.getTaskName(), dsName, dbName, tableName, tableDesc);
  }

  public PluginContext getPluginContext() {
    return pluginContext;
  }

  public void setPluginContext(PluginContext pluginContext) {
    this.pluginContext = pluginContext;
  }
}
